#ifndef __M_CHANNEL_H__
#define  __M_CHANNEL_H__
#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <sstream>
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_loger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include "muduo/net/TcpConnection.h"
#include "protobuf/codec/codec.h"
#include "protobuf/codec/dispatcher.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_route.hpp"

namespace mq{

//using是类型声明！
using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;//创建信道 句柄
using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;//关闭信道 句柄
using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;//创建交换机 句柄
using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;//关闭交换机句柄
using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;//创建队列句柄
using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;//关闭队列句柄
using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;//绑定句柄
using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;//解绑句柄
using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;//消息的发布句柄
using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;//消息确认的句柄
using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;//消费者订阅消息句柄
using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;//取消订阅句柄

class Channel{
private:
    std::string _cid;//信道id，唯一标识
    mq::Consumer::ptr _consumer;//消费者句柄

    mq::VirtualHost::ptr _host;//虚拟机管理句柄

    mq::ConsumerManager::ptr _cmp;//消费者管理句柄
    
    ProtobufCodecPtr _codec;//协议处理句柄
    muduo::net::TcpConnectionPtr _conn;//连接句柄
    mq::threadpool::ptr _pool;//任务池（线程池）
public:
    using ptr = std::shared_ptr<Channel>;

    Channel(const string& cid,
            const mq::VirtualHost::ptr& host,
            const mq::ConsumerManager::ptr& cmp,
            const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn,
            const mq::threadpool::ptr pool):_cid(cid),_host(host),
                _cmp(cmp),_codec(codec),_conn(conn),_pool(pool){
        DLOG("new Channel：%p ",this);//查看地址防止资源泄露，稳定服务器必须保证防止泄露

    }

    ~Channel(){
        //信道本质就是对应着一个消费者（消费者 -----信道-----  客户端）

        //所以若信道关闭了消费者就不存在了，那么也对应着他应该从消费者管理类中除去
    
        //1. 判断消费者是否存在
        if(_consumer.get() != nullptr){
            //存在则删除
            _cmp->remove(_consumer->tag,_consumer->qname);
        }
        DLOG("del Channel：%p ",this);//查看地址防止资源泄露

    }

//声明交换机
    void declareExchange(const declareExchangeRequestPtr& req){
        //调用虚拟机中的声明交换机函数
        DLOG("exchange_type：%d",req->exchange_type());    
        bool ret = _host->declareExchange(req->exchange_name(),req->exchange_type(),
                            req->durable(),req->auto_delete(),req->args());

        return basicResponse(ret,req->rid(),req->cid());
    }

//删除交换机
    void deleteExchange(const deleteExchangeRequestPtr& req){
        bool ret = _host->deleteExchange(req->exchange_name());
        return basicResponse(ret,req->rid(),req->cid());
    }

//声明队列
    void declareQueue(const declareQueueRequestPtr& req){
        bool ret = _host->declareQueue(req->queue_name(),req->durable(),req->exclusive(),req->auto_delete(),req->args());
        if(ret == false) {
            return basicResponse(false, req->rid(), req->cid());
        }
        _cmp->initQueueConsumer(req->queue_name());//初始化消费者管理中的消费者队列
        return basicResponse(ret,req->rid(),req->cid());
    }

//删除队列
    void deleteQueue(const deleteQueueRequestPtr& req){
        _cmp->destroyQueueConsumer(req->queue_name());
        bool ret = _host->deleteQueue(req->queue_name());
        return basicResponse(ret,req->rid(),req->cid());
    }

//交换机绑定队列
    void queueBind(const queueBindRequestPtr& req){
        bool ret = _host->bind(req->exchange_name(),req->queue_name(),req->binding_key());
        return basicResponse(ret,req->rid(),req->cid());
    }

//解除绑定
    void queueUnBind(const queueUnBindRequestPtr& req){
        bool ret = _host->unbind(req->exchange_name(),req->queue_name());
        return basicResponse(ret,req->rid(),req->cid());
    }

//发布消息:

//对于发布来说，我们需要找到该消息要发布到的队列，要找到队列那么就要借助交换机，通过交换机路由找到指定队列
//找到队列后就可以把消息添加进去，在通过队列名称知道对应的订阅者这样就能再将消息推送给订阅者消费了！
//而将最后这里推送给订阅者的操作单独出来为一个新函数 consumer
//其中还需要调用订阅者的回调函数callback（这个是在添加订阅时完成的），回调函数内部将发把 前面一直讲的消息给发布到指定的客户端
    void basicPublish(const basicPublishRequestPtr& req){
        //1. 获取交换机
        auto exchange = _host->selectExchange(req->exchange_name());
        
        BasicProperties* properties = nullptr;
        std::string routing_key;
        //获取属性中的值，为后面路由做准备
        if(req->mutable_properties() != nullptr){
            //判断 消息属性是否为空，若为空则表示广播交换
            // 反之从属性中获取properties和routing_key
            properties = req->mutable_properties();
            routing_key = properties->routing_key();
        }

        //获取所有指定交换机的绑定信息的情况
        auto mqbm = _host->exchangeBindings(exchange->name);
        //遍历绑定信息，查看是否有匹配路由的队列
        for(auto& binding : mqbm){
            //binding ：kv：交换机名称 : {队列名称 : binding_key} 
// 进行路由查看：
            // DLOG("查看绑定信息：type:%d，r：%s，b:%s",exchange->type,routing_key.c_str(),binding.second->binding_key.c_str());
            if(Router::route(exchange->type,routing_key,binding.second->binding_key)){
               //就代表找到了匹配的队列
                //将消息添加到队列中(发布消息)
                // DLOG("路由成功：r：%s，b:%s",routing_key.c_str(),binding.second->binding_key.c_str());
                _host->basicPublish(binding.second->msgqueue_name,properties,req->body());

                //将发布消息的函数打包丢进任务池（其中this指针让他获取消息，binding告诉他订阅的队列）
                // 实现发布消息时给订阅该队列的消费者发布信息的操作异步进行
                auto task = std::bind(&Channel::consum,this,binding.first);
                _pool->push(task);
            }
        }
        return basicResponse(true,req->rid(),req->cid());
    }

//消息确认
    void basicAck(const basicAckRequestPtr& req){
        _host->basicAck(req->queue_name(),req->message_id());
        return basicResponse(true,req->rid(),req->cid());
    }

//消息订阅
    void basicConsume(const basicConsumeRequestPtr& req){
        // DLOG("消息订阅，开始判断订阅的队列是否存在！");

        //判断消费者的订阅的队列是否存在
        if(!_host->existsQueue(req->queue_name())){
            DLOG("消息订阅失败，%s 消费者订阅队列不存在!", req->queue_name().c_str());
            return;
        }
        // DLOG("订阅的队列存在！，将callback回调函数适配");
        //在订阅队列中添加该新消费者
        auto cb = std::bind(&Channel::callback,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
        _consumer = _cmp->create(req->consumer_tag(),req->queue_name(),req->auto_ack(),cb);
        basicResponse(true,req->rid(),req->cid());
        // DLOG("给消费者；%s 订阅消息队列；%s 成功",req->consumer_tag().c_str(),req->queue_name().c_str());
    }

//取消订阅
    void basicCancel(const basicCancelRequestPtr& req){
        //从消费者队列中删除：
        _cmp->remove(req->consumer_tag(),req->queue_name());
        return basicResponse(true,req->rid(),req->cid());
    }

private:
//注意和消费者内部定义的回调函数，函数签名一致：
// using ConsumerCallback = std::function<void(const std::string&,const mq::BasicProperties* bp,const std::string&)>;
//内部主要就是执行调用发送函数，将消息发送给 客户端
    void callback(const std::string&tag,const mq::BasicProperties* bp,const std::string& body){
        //创建 消息的推送 响应 对象
        mq::basicConsumeResponse mbcr;
        mbcr.set_cid(_cid);
        mbcr.set_consumer_tag(tag);
        mbcr.set_body(body);
        if(bp != nullptr){
            mbcr.mutable_properties()->set_id(bp->id());
            mbcr.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            mbcr.mutable_properties()->set_routing_key(bp->routing_key());
        }
        _codec->send(_conn,mbcr);//basicConsumeResponse类型继承Message类型
        // DLOG("发回响应")
    }
//这是消息发布任务函数
//将从指定消息队列中获取一条消息，然后将消息发送回客户端
    void consum(const std::string& qname){
        //从指定队列获取消息
        mq::MessagePtr msg = _host->basicConsume(qname);
        if(msg.get() == nullptr){
            DLOG("执行消费任务失败，%s 队列没有消息!", qname.c_str());
            return ;
        }
        //从该订阅队列中获取一个消费者
        mq::Consumer::ptr mcp =  _cmp->choose(qname);
        if(mcp.get() == nullptr){
            DLOG("执行消费任务失败，%s 队列没有消息!", qname.c_str());
            return;
        }
        //调用订阅者的回调函数，并传递对应所需参数
        //执行回调消费者提前设置的回调函数
        mcp->callback(mcp->tag,msg->mutable_payload()->mutable_properties(),msg->payload().body());//调用callback，其中body就是消息！

        //判断是否自动确认，若是则直接确认消息
        if(mcp->auto_ack){
            //直接确认
            _host->basicAck(qname,msg->payload().properties().id());
        }
    }

    void basicResponse(bool ok,const std::string& rid,const std::string& cid){
        //class basicCommonResponse final : public ::PROTOBUF_NAMESPACE_ID::Message
        mq::basicCommonResponse resp;//通用响应
        resp.set_ok(ok);
        resp.set_rid(rid);
        resp.set_cid(cid);
        //void send(const muduo::net::TcpConnectionPtr& conn, const google::protobuf::Message& message)
        _codec->send(_conn,resp);//basicCommonResponse类型继承Message类型
    }

};

    class ChannelManager {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        bool openChannel(const std::string &id, 
            const VirtualHost::ptr &host, 
            const ConsumerManager::ptr &cmp, 
            const ProtobufCodecPtr &codec, 
            const muduo::net::TcpConnectionPtr &conn,
            const mq::threadpool::ptr &pool) {

            std::unique_lock<std::mutex> lock(_mutex);
            
            auto it = _channels.find(id);
            if (it != _channels.end()) {
                DLOG("信道：%s 已经存在!", id.c_str());
                return false;
            }
            auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }
        void closeChannel(const std::string &id){
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }
        
        Channel::ptr getChannel(const std::string &id) {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end()) {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };

}



#endif